package com.t11.consumer.controller;

import com.t11.common.pojo.LzwOrder;
import com.t11.common.pojo.LzwUser;
import com.t11.common.pojo.ResultMsg;
import com.t11.common.service.OrderService;
import com.t11.common.service.UserService;
import org.apache.dubbo.config.annotation.DubboReference;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.Map;

/**
 * 秒杀接口
 */
@RestController
public class SeckillController {

    @DubboReference
    UserService userService;

    @Resource
    RedisTemplate<String, Object> redisTemplate;

    @Resource
    OrderService orderService;

    @Resource
    KafkaTemplate<String, Object> kafkaTemplate;

    /**
     * 秒杀
     * @param params 商品id&用户名（手机号）
     * @return 秒杀结果
     */
    @PostMapping("seckill")
    @Transactional
    public ResultMsg doSeckill(@RequestBody Map<String, String> params) {
        String phone = params.get("phone");
        String goodsId = params.get("goodsId");
        String preemption = "preemption";
        
        // 判断用户是否为真
        LzwUser user = userService.findUserByPhone(phone);
        if (user == null) {
            return ResultMsg.USER_NULL;
        } else {
            // 判断该用户有没有订单
            LzwOrder order = orderService.findOrderByUserPhone(phone);

            synchronized (SeckillController.this) {
                HashOperations<String, Object, Object> opsForHash = redisTemplate.opsForHash();

                // 如果没有再判断是否预占
                if (order != null) {
                    // 如果订单有信息，表示已购买成功
                    return ResultMsg.IS_BUY;
                } else {
                    boolean flag = opsForHash.hasKey(preemption, phone);
                    if (flag) {
                        // 如果预占集合有信息表示已预占
                        return ResultMsg.IS_PREEMPTION;
                    }
                }

                // 又没订单又没预占才会执行下面
                String key = "seckill:stock:" + goodsId;
                // 预占库存
                ValueOperations<String, Object> opsForValue = redisTemplate.opsForValue();
                // 前提库存大于0
                Integer stock = (Integer) opsForValue.get(key);
                if (stock == null || stock <= 0) {
                    return ResultMsg.STOCK_NULL;
                } else {
                    // 原子性操作 预占库存减1
                    opsForValue.decrement(key);
                    // redis加入预占集合
                    opsForHash.put(preemption, phone, "");

                    // 预占成功后转入支付操作
                    boolean flag = orderService.payMoney();

                    if (flag) {
                        // 支付成功
                        System.out.println("支付成功：" + phone);
                        // 支付成功后排到kafka消息队列进行并发排队
                        LzwOrder lzwOrder = new LzwOrder();
                        lzwOrder.setUserPhone(phone);
                        lzwOrder.setGoodsId(Integer.valueOf(goodsId));
                        kafkaTemplate.send("topic-lzw", lzwOrder);
                        return ResultMsg.SECKILL_SUCCESS;
                    } else {
                        // 支付失败
                        System.out.println("支付失败：" + phone);
                        // 库存回滚
                        opsForValue.increment(key);
                        // 支付失败删除预占集合
                        opsForHash.delete(preemption, phone);
                        return ResultMsg.PAY_FAIL;
                    }
                }
            }
        }
    }
}
